﻿#region License
/* ---------------------------------------------------------------- *
 *
 * This file is part of the Xcoordination Application Space
 * ("XcoAppSpace") http://www.xcoordination.com
 *
 * Copyright (C) 2009 Xcoordination http://www.xcoordination.com
 *
 * XcoAppSpace is free software; you can redistribute it and/or
 * modify it under the terms of version 2 of the GNU General
 * Public License as published by the Free Software Foundation.
 *
 * XcoAppSpace is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program. If not, see http://www.gnu.org/licenses/
 * or write to the Free Software Foundation, Inc., 51 Franklin
 * Street, Fifth Floor, Boston, MA 02110-1301, USA.
 *
 * ---------------------------------------------------------------- */
#endregion

using System;
using System.Linq;
using System.Collections.Generic;
using Microsoft.Ccr.Core;
using XcoAppSpaces.Contracts.Communication;
using XcoAppSpaces.Contracts.Messaging;
using XcoAppSpaces.Contracts.Transport;
using XcoAppSpaces.Contracts.Worker;
using XcoAppSpaces.Core.Communication.WorkerResolve;
using XcoAppSpaces.Contracts.Logging;
using XcoAppSpaces.Contracts.Exceptions;
using XcoAppSpaces.Contracts.Security;
using XcoAppSpaces.Contracts.Serialization;
using XcoAppSpaces.Contracts.Service;

namespace XcoAppSpaces.Core.Communication
{
	class MessageTransmitter : IDisposable
	{
		private readonly IXcoLogger log = XcoLoggerFactory.CreateInstance(typeof(MessageTransmitter));
		private readonly IXcoWorkerRegistry localWorkers;
		private readonly IXcoSecurityService securityService;
		private readonly CommunicationEvents commEvents;
		private readonly Dictionary<string, XcoServiceConfiguration<IXcoTransportService>> commServices = new Dictionary<string, XcoServiceConfiguration<IXcoTransportService>>();
		private readonly Dictionary<string, IXcoSerializer> serializers = new Dictionary<string, IXcoSerializer>();
		private string defaultCommServiceName;
		private RemotePortRegistry portRegistry;

		public Port<XcoCommunicationError> Errors { get; private set; }
		private readonly Guid errorPortID = new Guid("44c362d9-87b2-4863-b15e-8caed4594462");

		public MessageTransmitter(IXcoServiceRegistry serviceRegistry, CommunicationEvents commEvents)
		{
			this.localWorkers = serviceRegistry.Resolve<IXcoWorkerRegistry>();
			this.securityService = serviceRegistry.Resolve<IXcoSecurityService>();
			InitTransportServicesAndSerializers(serviceRegistry);
            this.Errors = new Port<XcoCommunicationError>();
			this.commEvents = commEvents;
		}

		private void InitTransportServicesAndSerializers(IXcoServiceRegistry serviceRegistry)
		{
			var registeredCommServices = serviceRegistry.ResolveAll<IXcoTransportService>();
			var registeredSerializers = serviceRegistry.ResolveAll<IXcoSerializer>();
			var defaultSerializer = registeredSerializers.FirstOrDefault(s => s.IsDefault);
			if (defaultSerializer == null)
				throw new XcoConfigurationException("No default serializer found.");
			foreach (var commService in registeredCommServices)
			{
				commService.Service.OnMessageReceived += MessageReceived;
				this.commServices.Add(commService.Name, commService);
				if (commService.IsDefault)
					defaultCommServiceName = commService.Name;
				if (commService.Mappings == null || commService.GetMapping<IXcoSerializer>() == null)
					this.serializers.Add(commService.Name, defaultSerializer.Service);
				else
				{
					var serializer = registeredSerializers.FirstOrDefault(s => s.Name == commService.GetMapping<IXcoSerializer>());
					if (serializer == null)
						throw new XcoConfigurationException("No serializer with name " + commService.GetMapping<IXcoSerializer>() + " found.");
					this.serializers.Add(commService.Name, serializer.Service);
				}
			}
		}

		public string LocalAddress
		{
			get { return (commServices.Count > 0 ? commServices[defaultCommServiceName].Service.LocalAddress : null); }
		}

		public IEnumerable<XcoServiceConfiguration<IXcoTransportService>> CommServiceInfo
		{
			get { return commServices.Values; }
		}

		public void Start()
		{
			portRegistry = new RemotePortRegistry();
			portRegistry.RegisterPermanently(this.Errors, this.errorPortID);
		}

		public XcoServiceConfiguration<IXcoTransportService> GetCommServiceInfo(string commServiceName)
		{
			return commServices[CheckCommServiceName(commServiceName)];
		}

		public IXcoSerializer GetSerializer(string commServiceName)
		{
			return serializers[commServiceName];
		}

		public void SendMessage(MessageToSend msg)
		{
			try
			{
				var commServiceName = CheckCommServiceName(msg.CommServiceName);
				var commService = commServices[commServiceName];
				RemoveAckCoordinationPortsBeforeSend(msg.Context);
				log.Debug("Sending " + msg.Data.GetType() + " to \"" + msg.RemoteAddress + "\", using CommService: " + commService.Service.LocalAddress);
				using (new SerializationContext(portRegistry.Register, SendMessage, commService.Service.LocalAddress, commService.Name))
				{
					var xm = new XcoMessage
					          	{
					          		CausalityContext = serializers[commServiceName].SerializeCausalityContext(msg.Context),
					          		Content = serializers[commServiceName].SerializeMessageContent(msg.Data),
					          		ContentInfo = msg.Data.ToString()
					          	};
					commEvents.RaiseOnSendingMessageEvent(xm, msg.RemoteAddress, commServiceName);
				
					var ackPort = msg.CommAckPort ?? GetAckPort();
					if (ackPort != null)
						xm.OnTransferSuccessful += _=> RespondAck(ackPort);

					xm.OnTransferFailed += (_, ex) => HandleExceptionDuringSend(msg, ex);

					commService.Service.Send(xm, msg.RemoteAddress);
					//Console.WriteLine("Message Length: " + serializer.Serialize(msg).Length);
				}
			}
			catch (Exception ex)
			{
				HandleExceptionDuringSend(msg, ex);
			}
		}

		private string CheckCommServiceName(string commServiceName)
		{
			if (commServices.Count == 0)
			{
				log.Error("Error sending message to remote port or worker: There are no registered transport services.");
				throw new XcoCommunicationException("There are no registered transport services.");
			}
			if (commServiceName == null)
				commServiceName = defaultCommServiceName;
			if (!commServices.ContainsKey(commServiceName))
			{
				log.Error("Error sending message to remote port or worker: The transport service \"" + commServiceName + "\" does not exist.");
				throw new XcoCommunicationException("The transport service \"" + commServiceName + "\" does not exist.");
			}
			return commServiceName;
		}

		private static void RemoveAckCoordinationPortsBeforeSend(CausalityContext context)
		{
			if (context == null)
				return;
			foreach (var c in context.Causalities)
			{
				if (c.CoordinationPort != null && c.CoordinationPort is Port<Ack>)
					c.CoordinationPort = null;
			}
			//remove any causalities that don't have any ports  (e.g. because the ack port was removed)
			if (context.Causalities.Any(c => c.ExceptionPort == null && c.CoordinationPort == null))
				context.Causalities = context.Causalities.Where(c => c.ExceptionPort != null || c.CoordinationPort != null).ToArray();
		}

		private static Port<Ack> GetAckPort()
		{
			if (!Dispatcher.HasActiveCausalities)
				return null;
			var c = Dispatcher.ActiveCausalities.First();
			if (c.CoordinationPort != null && c.CoordinationPort is Port<Ack>)
				return (Port<Ack>)c.CoordinationPort;
			return null;
		}

		private static void RespondAck(Port<Ack> ackPort)
		{
			if (ackPort != null)
				ackPort.Post(new Ack(DateTime.Now));
		}

		private void HandleExceptionDuringSend(MessageToSend msg, Exception ex)
		{
			if (msg.CommErrorPort != null)
			{
				msg.CommErrorPort.Post(ex); //CommErrorPort is used for handling calls to port.Send()
			}
			else
			{
				if (msg.Context == null || msg.Context.Causalities.Length == 0 || msg.Context.Causalities[0].ExceptionPort == null)
					Errors.Post(new XcoCommunicationError { Error = ex, Info = "Error sending message to " + msg.RemoteAddress, SourceAddress = this.LocalAddress });
				else
					msg.Context.Causalities[0].ExceptionPort.PostUnknownType(ex);
			}
		}

		public void MessageReceived(XcoMessage msg, string remoteAddress, IXcoTransportService commService)
		{
			string commServiceName = GetCommServiceName(commService);
			try
			{
				commEvents.RaiseOnMessageReceivedEvent(msg, remoteAddress, commServiceName);
				using (new SerializationContext(portRegistry.Register, SendMessage, commServices[commServiceName].Service.LocalAddress, commServiceName))
				{
					CausalityContext context;
					RemoteMessage content;
					if (!TryDeserializeMessage(commServiceName, msg, remoteAddress, out context, out content))
						return;

					context.InstallCausalities(CreateMessageContext(remoteAddress, content, commServiceName));
					//interpret message content
					if (content is WorkerMessage)
					{
						//worker message: find worker and post message to its portset
						ProcessWorkerMessage((WorkerMessage)content, remoteAddress, msg, context, commServiceName);
					}
					else if (content is RemotePortMessage)
					{
						//remote port message: find port to given id and post message to it
						ProcessRemotePortMessage((RemotePortMessage)content);
					}
					else
					{
						string errorMsg = (content != null) ?
							("MessageTransmitter: ERROR unknown message content received: " + content) :
							"MessageTransmitter: ERROR message content could not be deserialized!";
						log.Equals(errorMsg);
						RespondErrorToCausalityContext(msg, remoteAddress, errorMsg, null, context, commServiceName);
					}
					context.UninstallCausalities();
				}
			}
			catch (Exception ex)
			{
				log.Error("Error handling received message.", ex);
			}
		}

		private MessageContext CreateMessageContext(string remoteAddress, RemoteMessage content, string transportServiceName)
		{
			var context = new MessageContext
			       	{
			       		IsRemoteMessage = true,
			       		RemoteAddress = remoteAddress,
						TransportServiceName = transportServiceName
			       	};
			if (content is WorkerMessage)
			{
				var sessionID = ((WorkerMessage) content).SessionID;
				context.SessionID = sessionID;
				context.Username = securityService.GetUsername(sessionID);
			}
			return context;
		}

		private bool TryDeserializeMessage(string commServiceName, XcoMessage msg, string remoteAddress, out CausalityContext context, out RemoteMessage content)
		{
			context = null;
			content = null;
			//first deserialize causality context, if content deserialization fails post response to exception port
			try
			{
				context = serializers[commServiceName].DeserializeCausalityContext(msg.CausalityContext);
			}
			catch (Exception ex)
			{
				log.Error("Error deserializing causalities of received message.", ex);
				//respond error
				RespondErrorToCausalityContext(msg, remoteAddress, "Error deserializing causalities of received message.", ex, null, commServiceName);
				return false;
			}
			try
			{
				//deserialize content
				content = serializers[commServiceName].DeserializeMessageContent(msg.Content);
			}
			catch (Exception ex)
			{
				log.Error("Error deserializing content of received message.", ex);
				//respond error
				RespondErrorToCausalityContext(msg, remoteAddress, "Error deserializing content of received message.", ex, context, commServiceName);
				return false;
			}
			return true;
		}

		private void ProcessWorkerMessage(WorkerMessage message, string remoteAddress, XcoMessage originalMsg, CausalityContext context, string commServiceName)
		{
			//worker message: find worker and post message to its portset
			var worker = localWorkers.Resolve(message.WorkerIdentifier, true);
			if (worker != null)
			{
				log.Debug("Message for worker received: " + message.WorkerIdentifier + ", on CommService: " + commServices[commServiceName].Service.LocalAddress);
				//check permission to access this worker port (if worker is resolveworker, permission is always granted)
				if (!(worker.Worker is ResolveWorker))
				{
					try
					{
						securityService.CheckPermission(worker.Worker, message.Msg, remoteAddress, message.SessionID);
					}
					catch (Exception ex)
					{
						log.Warn("Permission request to access worker failed: " + ex.Message);
						RespondErrorToCausalityContext(originalMsg, remoteAddress, "Permission request to access worker failed.", ex, context, commServiceName);
						return;
					}
				}
				//post message to worker
				worker.Worker.PostUnknownType(message.Msg);
			}
			else
			{
				log.Debug("Message for unknown or unpublished worker received (is ignored)! Identifier = " + message.WorkerIdentifier + ", on CommService: " + commServices[commServiceName].Service.LocalAddress);
			}
		}

		private void ProcessRemotePortMessage(RemotePortMessage message)
		{
			//remote port message: find port to given id and post message to it
			var port = portRegistry.Resolve(message.PortID);
			if (port != null)
				port.PostUnknownType(message.Msg);
			else
				log.Warn("Message for unknown remote port is ignored! ID: " + message.PortID +
						", message type: " + (message.Msg != null ? message.Msg.GetType().ToString() : "NULL"));
		}

		private string GetCommServiceName(IXcoTransportService commService)
		{
			foreach (var val in commServices)
			{
				if (val.Value.Service == commService)
				{
					return val.Key;
				}
			}
			return null;
		}

		private void RespondErrorToCausalityContext(XcoMessage msg, string remoteAddress, string info, Exception exception, CausalityContext causalityContext, string commServiceName)
		{
			//respond to sender of the message with an error because message content could not be deserialized
			if (causalityContext != null && 
				causalityContext.Causalities != null &&
                causalityContext.Causalities.Length > 0 &&
				causalityContext.Causalities[0].ExceptionPort != null)
			{
				if (causalityContext.Causalities[0].ExceptionPort.TryPostUnknownType(exception))
					return;
			}
			//respond to space's uncaught exception port if no valid causality exception port was available
			info += " Further message information: " + (msg.ContentInfo ?? "[not available]");
			SendUncaughtCommunicationError(remoteAddress, commServiceName,
			                               new XcoCommunicationError(exception, info, commServices[commServiceName].Service.LocalAddress));
		}

		private void SendUncaughtCommunicationError(string spaceAddress, string commServiceName, XcoCommunicationError error)
		{
			this.SendMessage(new MessageToSend(spaceAddress, commServiceName, null, new RemotePortMessage(errorPortID, error), null, null));
		}

		#region IXcoService Members

		public void Initialize(IXcoServiceRegistry serviceRegistry)
		{
			//nothing to do
		}

		#endregion

		public void Dispose()
		{
			portRegistry.Dispose();
		}
	}
}
